package org.big.kafkacluster;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@RestController
@RequestMapping("/kafka")
public class KafkaTestController {

    @Autowired
    private KafkaProducerService kafkaProducerService;

    private Random random = new Random();
    private ExecutorService executorService = Executors.newFixedThreadPool(10);

    //localhost:8080/kafka/send?message=this is a message
    @GetMapping("/send")
    public String sendTestMessage(@RequestParam String message) {
        kafkaProducerService.sendMessage(message);
        return "消息 '" + message + "' 已发送！";
    }


    //localhost:8080/kafka/send?message=this is a message
    @GetMapping("/test")
    public String test() throws InterruptedException {
        extracted();
        return "消息 '" + " test " + "' 已发送！";
    }



    @Scheduled(fixedRate = 60000)
     void extracted() throws InterruptedException {
        for (int i = 0; i < 10; i++) {
            int randomValue = new Random().nextInt(100);
            Thread.sleep(30);
            String message = "random String：" + randomValue + " (Task: " + i + ")";
            kafkaProducerService.sendMessage(message);
            System.out.println(Thread.currentThread().getName() + " 发送消息: " + message);

        }
    }
}